# coding:utf-8
"""
Code that goes along with the Airflow tutorial located at:
https://github.com/airbnb/airflow/blob/master/airflow/example_dags/tutorial.py
"""
from datetime import datetime, timedelta

from airflow import DAG
from airflow.operators.bash_operator import BashOperator

default_args = {
    'owner': 'apps',
    'depends_on_past': False,
    'start_date': datetime(2018, 11, 6, 1, 0),
    'email': ['youfeng@stargraph.cn', 'zhangjun@stargraph.cn', 'zhoujiamu@stargraph.cn', 'zhujiangbao@stargraph.cn'],
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 0,
    'max_active_runs': 1,
    'retry_delay': timedelta(minutes=30),
    # 'queue': 'bash_queue',
    # 'pool': 'backfill',
    # 'priority_weight': 10,
    # 'end_date': datetime(2016, 1, 1),
}
dag = DAG('inc_data_v3', max_active_runs=1, default_args=default_args, schedule_interval='0 1 * * *')

APPS_DIR = "/home/dig/airflow/apps/"
# t1, t2 and t3 are examples of tasks created by instatiating operators
inc_parser = BashOperator(
    task_id='inc_parser',
    bash_command='source ~/.bash_profile; cd {}/graph-builder/spark_processor/bin; sh inc_parser.sh; '.format(APPS_DIR),
    dag=dag)

inc_process = BashOperator(
    task_id='inc_process',
    bash_command='source ~/.bash_profile; cd {}/graph-builder/spark_processor/bin; sh inc_process.sh; '.format(
        APPS_DIR),
    retries=0,
    dag=dag)

filter_changsha_image = BashOperator(
    task_id='filter_changsha_image',
    bash_command='source ~/.bash_profile; cd {}/graph-builder/spark_processor/bin; sh inc_changsha.sh; '.format(
        APPS_DIR),
    retries=0,
    dag=dag)

# 算法和tag都需要依赖这个
image_to_hbase = BashOperator(
    task_id='image_to_hbase',
    bash_command='source ~/.bash_profile && cd {}/graph-utils/image2h-scala && sh run.sh; '.format(APPS_DIR),
    retries=0,
    dag=dag)

copy_local = BashOperator(
    task_id='copy_local',
    bash_command='source ~/.bash_profile; cd {}/graph-builder/; sh copy_local.sh; '.format(APPS_DIR),
    retries=0,
    dag=dag)

copy_local_with_changsha = BashOperator(
    task_id='copy_local_with_changsha',
    bash_command='source ~/.bash_profile; cd {}/graph-builder/; sh copy_local_with_changsha.sh; '.format(APPS_DIR),
    retries=0,
    dag=dag)

upload_to_lz = BashOperator(
    task_id='upload_to_lz',
    bash_command='source ~/.bash_profile; cd {}/upload; sh tar_and_upload.sh; '.format(APPS_DIR),
    retries=0,
    dag=dag)

upload_to_changsha = BashOperator(
    task_id='upload_to_changsha',
    bash_command='source ~/.bash_profile; cd {}/upload; sh upload_basic_data.sh; sh upload_mining_data.sh; sh upload_label_data.sh; sh upload_jtpx_data.sh; '.format(
        APPS_DIR),
    retries=0,
    dag=dag)

basedata_to_hive = BashOperator(
    task_id='basedata_to_hive',
    bash_command='source ~/.bash_profile; cd /data/dfs/dn/1/ware/bigdata/etl_dap/ODM/shell; sh hive_load_zstp_arango_increment.sh; sh hive_load_zstp_mongo_increment.sh; sh hive_copy_local_zstp.sh; ',
    retries=0,
    dag=dag)

upload_hive_to_changsha = BashOperator(
    task_id='upload_hive_to_changsha',
    bash_command='source ~/.bash_profile; cd {}/upload; sh upload_hive_data.sh; '.format(APPS_DIR),
    retries=0,
    dag=dag)

# dump_mongo = BashOperator(
#     task_id='dump_mongo',
#     bash_command='source ~/.bash_profile; cd {}; sh mongo_dump.sh; '.format(APPS_DIR),
#     retries=0,
#     dag=dag)

clean_up = BashOperator(
    task_id='clean_up',
    bash_command='source ~/.bash_profile; cd {}; sh cleanup.sh; '.format(APPS_DIR),
    retries=0,
    dag=dag)
correct_shareholder = BashOperator(
    task_id='correct_shareholder',
    bash_command='source ~/.bash_profile;cd %s/graph_correct/graph-tools/bin;sh collect_version.sh|xargs -i sh correct_shareholder.sh {};' % APPS_DIR,
    retries=0,
    dag=dag
)
correct_contributor = BashOperator(
    task_id='correct_contributor',
    bash_command='source ~/.bash_profile;cd %s/graph_correct/graph-tools/bin;sh collect_version.sh|xargs -i sh correct_contributor.sh {};' % APPS_DIR,
    retries=0,
    dag=dag
)
correct_annualreport_invest = BashOperator(
    task_id='correct_annualreport_invest',
    bash_command='source ~/.bash_profile;cd %s/graph_correct/graph-tools/bin;sh collect_version.sh|xargs -i sh correct_annualreport_invest.sh {};' % APPS_DIR,
    retries=0,
    dag=dag
)
correct_changerecord_invest = BashOperator(
    task_id='correct_changerecord_invest',
    bash_command='source ~/.bash_profile;cd %s/graph_correct/graph-tools/bin;sh collect_version.sh|xargs -i sh correct_changerecords_invest.sh {};' % APPS_DIR,
    retries=0,
    dag=dag
)
correct_officer = BashOperator(
    task_id='correct_officer',
    bash_command='source ~/.bash_profile;cd %s/graph_correct/graph-tools/bin;sh collect_version.sh|xargs -i sh correct_officer.sh {};' % APPS_DIR,
    retries=0,
    dag=dag
)
correct_belong = BashOperator(
    task_id='correct_belong',
    bash_command='source ~/.bash_profile;cd %s/graph_correct/graph-tools/bin;sh collect_version.sh|xargs -i sh correct_belong.sh {};' % APPS_DIR,
    retries=0,
    dag=dag
)
correct_legalman = BashOperator(
    task_id='correct_legalman',
    bash_command='source ~/.bash_profile;cd %s/graph_correct/graph-tools/bin;sh collect_version.sh|xargs -i sh correct_legalman.sh {};' % APPS_DIR,
    retries=0,
    dag=dag
)
correct_finish = BashOperator(
    task_id='correct_finish',
    bash_command='source ~/.bash_profile;cd %s/upload; sh correct_finish.sh; ' % APPS_DIR,
    retries=0,
    dag=dag
)
imp_arango = BashOperator(
    task_id='import_arango',
    bash_command='source ~/.bash_profile; cd {}/graph-builder/; sh imp.sh; '.format(APPS_DIR),
    retries=0,
    dag=dag)

inc_parser.set_downstream(inc_process)
inc_process.set_downstream(copy_local)
inc_process.set_downstream(filter_changsha_image)

copy_local.set_downstream(correct_officer)
correct_officer.set_downstream(correct_belong)
correct_belong.set_downstream(correct_contributor)
correct_contributor.set_downstream(correct_shareholder)
correct_shareholder.set_downstream(correct_legalman)
correct_legalman.set_downstream(correct_changerecord_invest)
correct_changerecord_invest.set_downstream(correct_annualreport_invest)
correct_annualreport_invest.set_downstream(correct_finish)

# copy_local.set_downstream(dump_mongo)
# dump_mongo.set_downstream(upload_to_lz)

# correct_finish.set_upstream(correct_officer)
# correct_finish.set_upstream(correct_belong)
# correct_finish.set_upstream(correct_contributor)
# correct_finish.set_upstream(correct_shareholder)
# correct_finish.set_upstream(correct_legalman)
# correct_finish.set_upstream(correct_changerecord_invest)
# correct_finish.set_upstream(correct_annualreport_invest)

correct_finish.set_downstream(upload_to_lz)
correct_finish.set_downstream(imp_arango)
correct_finish.set_downstream(image_to_hbase)
correct_finish.set_downstream(copy_local_with_changsha)
copy_local_with_changsha.set_downstream(upload_to_changsha)
copy_local_with_changsha.set_downstream(basedata_to_hive)

upload_hive_to_changsha.set_upstream(upload_to_changsha)
upload_hive_to_changsha.set_upstream(basedata_to_hive)

copy_local_with_changsha.set_upstream(filter_changsha_image)
imp_arango.set_downstream(clean_up)
image_to_hbase.set_downstream(clean_up)
upload_to_lz.set_downstream(clean_up)
